#include "xworker_priority_queue.h"
#include "xthread_manager.h"
#include "xdebug.h"

namespace xge{
namespace xge_internal{

  const char_t* WorkerPriorityQueue::kAboutToQuit = _xT("quit");
  const char_t* WorkerPriorityQueue::kQuitEmpty   = _xT("quit_empty");
  // Dequeue
  size_t WorkerPriorityQueue::Dequeue( TaskPtrVector* container , size_t size  ) {
    // clear the container
    container->clear();
    // get the specific size
    const size_t qsize = queue_.size();
    if( size ==0 ) {
      size = qsize; 
    } else {
      size = qsize < size ? qsize : size;
    }
    // preallocating the size of this container
    if( container->capacity() < size ) {
      container->reserve(size);
    }
    size_t i = 0; // counter
    queue_sync_lock_.Lock();
    for( i = 0 ; i < size ; ++i ) {
      queue_.top().task_ptr->set_task_status( Task::TASK_STATUS_LOCKING );
      container->push_back( TaskPtr(queue_.top().task_ptr) );
      queue_.pop();
    }
    queue_sync_lock_.Unlock();
    return i;
  }

  bool WorkerPriorityQueue::PushTask( const TaskPtr& task_ptr ) {
    //Test our status is able to push task
    if( status() == WorkerQueue::STATUS_RELEASE || status() == WorkerQueue::STATUS_LOCKING ) {
      return false;
    }
    PriorityTask* down_cast_ptr = task_ptr->DownCast<PriorityTask>();
    if( down_cast_ptr->priority() > cur_max_priority_ ) {
      cur_max_priority_ = down_cast_ptr->priority();
    }

    task_ptr->set_task_status( Task::TASK_STATUS_WAITING );
    // Nothing to check , the caller will check 
    queue_sync_lock_.Lock();
    queue_.push(  SPriorityProxy(task_ptr) );
    queue_sync_lock_.Unlock();
    ++enqueue_times_totally_;
    // wake up the back thread
    WakeUp();
    return true;
  }

  bool WorkerPriorityQueue::PushMessage( const TaskPtr& task_ptr ) {
    //Test our status is able to push task
    if( status() == WorkerQueue::STATUS_RELEASE || status() == WorkerQueue::STATUS_LOCKING ) {
      return false;
    }
    // The difference between push message and push task is
    // the push message will definitely set the XWorkerPriorityQueue to
    // Locking status , and it must be placed here , otherwise 
    // it may lead to the inconsistent problem
    set_status( WorkerQueue::STATUS_LOCKING );
    // Nothing to check , the caller will check 
    queue_sync_lock_.Lock();
    queue_.push(  SPriorityProxy(task_ptr) );
    queue_sync_lock_.Unlock();
    // Wake up the back thread 
    WakeUp();
    return true;

  }

  size_t WorkerPriorityQueue::PushTask( TaskPtrList* task_ptr_list ) {
    //Test our status is able to push task
    if( status() == WorkerQueue::STATUS_RELEASE || status() == WorkerQueue::STATUS_LOCKING ) {
      return 0;
    }
    // using a count to record the size
    // instead of using task_ptr_list->size()
    // since many realization of list::size() is
    // time costing , they just use std::difference( begin() ,end() ) which tasks O(n) time ( VS do not use this )    
    size_t num = 0;
    // Batch operation 
    // Since this internal queue can not push batch internally 
    // like using split or swap operation 
    // so we just handle one by one 
    // Here in fact , we can do like this two
    // Lock every single push operation 
    // however , this may make the back thread wake up and make this PushTask
    // very time costing 
    // So we choose to Lock the whole process , so the back thread will wake        
    queue_sync_lock_.Lock();
    for( TaskPtrList::iterator ib = task_ptr_list->begin() , ie = task_ptr_list->end() ; ib != ie ; ++ib , ++num ) {
      // check the max priority here
      PriorityTask* down_cast_ptr = (*ib)->DownCast<PriorityTask>();
      if( down_cast_ptr->priority() > cur_max_priority_ ) {
        cur_max_priority_ = down_cast_ptr->priority();
      }
      queue_.push( SPriorityProxy((*ib)) );
    }
    queue_sync_lock_.Unlock();
    // Now wake up the back thread if need
    WakeUp();
    enqueue_times_totally_+= num;
    return num;
  }

  bool WorkerPriorityQueue::TryPushTask( const TaskPtr& task_ptr ) {
    //Test our status is able to push task
    if( status() == WorkerQueue::STATUS_RELEASE || status() == WorkerQueue::STATUS_LOCKING ) {
      return false;
    }
    // Update the max priority now
    PriorityTask* down_cast_ptr = task_ptr->DownCast<PriorityTask>();
    if( down_cast_ptr->priority() > cur_max_priority_ ) {
      cur_max_priority_ = down_cast_ptr->priority();
    }
    // Update the task status
    task_ptr->set_task_status( Task::TASK_STATUS_WAITING );
    bool try_lock_ret = queue_sync_lock_.TryLock();
    if( !try_lock_ret ) {
      return false;
    }
    queue_.push(  SPriorityProxy(task_ptr) );
    queue_sync_lock_.Unlock();
    ++enqueue_times_totally_;
    // wake up the back thread
    WakeUp();
    return true;
  }


  void WorkerPriorityQueue::ClearQueue() {
    while( !queue_.empty() ) {
      queue_.pop();
    }
  }

  int WorkerPriorityQueue::Loop() {
    // Update the status to working
    set_status( WorkerQueue::STATUS_WORKING );

    struct AutoReleaseTaskContainer {
      AutoReleaseTaskContainer(TaskPtrVector* ptr) : container_ptr(ptr) {}

      ~AutoReleaseTaskContainer() {
        for( TaskPtrVector::iterator ib = container_ptr->begin() , ie = container_ptr->end() ; ib != ie ; ++ib ) {
          (*ib)->set_task_status( Task::TASK_STATUS_RELEASE );
        }
      }
      TaskPtrVector* container_ptr;
    };
    // the task dealing loop 
    // the task cache container
    TaskPtrVector task_cache_container ;
    // This decalaration MUST be behind the XPriorityTaskPtrVector task_cache_container;
    // since this will make the DTor call before the task_cache_conatiner dtor call 
    AutoReleaseTaskContainer releaser(&task_cache_container);
    // Begin the task loop
    while(true) {
      if( empty() ) {
        // This must be atomic 
        // since the push operation will wake up the thread by query the status
        set_status_atomic( WorkerQueue::STATUS_SLEEP );
        waitable_event_.Wait();
        // This status will not be query by other place
        // so it is a released store 
        set_status( WorkerQueue::STATUS_WORKING );
      } 
      // The original version of loop has been deleted 
      // this is the new version of loop
      // When reaching here there must be existed the task inside the queue 
      Dequeue(&task_cache_container,precache_size_);
      for( TaskPtrVector::iterator ib = task_cache_container.begin() , ie = task_cache_container.end() ; ib != ie ; ++ib ) {
        // This is a remote binding task not a message
        if( (*ib)->bind_type() == REMOTE_BINDING ) {
          (*ib)->set_task_status( Task::TASK_STATUS_RUNNING );
          (*ib)->Execute();
          (*ib)->set_task_status( Task::TASK_STATUS_RELEASE );
          continue;
        } else if( (*ib)->bind_type() == REMOTE_MESSAGE ) {
          // Now it is message task not a real task
          // Anyway this message will be deleted , so set its status to release
          (*ib)->set_task_status( Task::TASK_STATUS_RELEASE );
          PriorityTask* priority_task_ptr = (*ib)->DownCast<PriorityTask>();
          if( priority_task_ptr->name() == kAboutToQuit ) {
            // now we have to quit now
            set_status_atomic( WorkerQueue::STATUS_RELEASE );
            return LOOP_OK_RETURN;
          } else if( priority_task_ptr->name() == kQuitEmpty ) {
            //1. Execute the task in the task_cache_container
            for( ++ib ; ib != ie ; ++ib ) {
              (*ib)->set_task_status( Task::TASK_STATUS_RUNNING );
              (*ib)->Execute();
              (*ib)->set_task_status( Task::TASK_STATUS_RELEASE );
            }// end for
            //2. Execute the task in the queue
            Dequeue(&task_cache_container,0);
            for( ib = task_cache_container.begin() , ie = task_cache_container.end() ; ib != ie ; ++ib ) {
              (*ib)->set_task_status( Task::TASK_STATUS_RUNNING );
              (*ib)->Execute();
              (*ib)->set_task_status( Task::TASK_STATUS_RELEASE );
            }// end for
            //3. set the status and quit
            // this must be the atomic 
            set_status_atomic( WorkerQueue::STATUS_RELEASE );
            return LOOP_OK_RETURN;
          } else {
            // when reaching here
            // this must the message we can not handle
            SYS_LOG(PANIC,false,_xT("XWorkerPriority meet unhandle message in the loop!Message: %s.\n"),(priority_task_ptr->name()).c_str());
            // simply omit this message 
            // go on running
            continue;
          }// end if-else
        }// end if-else
      }//end for
    } // end while
    // impossible branch
    UNREACH_BRANCH(_xT("XWorkerPriority::Loop can not reach out of while loop!"));
    return LOOP_FAILED_RETURN;
  }


  bool WorkerPriorityQueue::WaitWhenRelease() {
    // the caller will validate the calling environment 
    while( status_atomic()  != STATUS_RELEASE ) {
      ThreadManager::SwithToAnotherThread();
    }
    return true;
  }

  const TripleResult WorkerPriorityQueue::WaitWhenRelease( int ms ) {
    // avoid calling timer
    // so test the empty at first
    if( status_atomic()  == STATUS_RELEASE ) {
      return TripleResult::kSucceded;
    }

    Timer timer;
    Timer::tick_t time_diff;
    if( !timer.BeginTimerCounting() ) {
      return TripleResult::kException;
    }
    while( status()  != STATUS_RELEASE ) {
      time_diff+=timer.TimeDiff();
      if( time_diff >= (size_t)(ms) ) {
        return TripleResult::kFailed;
      }
    }
    return TripleResult::kSucceded;
  }

  bool WorkerPriorityQueue::WaitWhenEmpty() {
    // the caller will validate the calling environment 
    while( !empty() ) {
      ThreadManager::SwithToAnotherThread();
    }
    return true;
  }

  const TripleResult WorkerPriorityQueue::WaitWhenEmpty( int ms ) {
    // avoid calling timer
    // so test the empty at first
    if( empty() ) {
      return TripleResult::kSucceded;
    }
    Timer timer;
    Timer::tick_t time_diff;
    if( !timer.BeginTimerCounting() ) {
      return TripleResult::kException;
    }
    while( !empty() ) {
      time_diff+=timer.TimeDiff();
      if( time_diff >= (size_t)(ms) ) {
        return TripleResult::kFailed;
      }
    }
    return TripleResult::kSucceded;
  }

}// namespace xge_internal
}// namespace xge